package com.whz.mlc.configuration;

import com.whz.generic.threads.DefaultCommonThreadPool;
import com.whz.mlc.cache.second.redis.MlcConstants;
import com.whz.mlc.cache.second.redis.client.MlcRedisClient;
import com.whz.mlc.cache.second.redis.listener.ClusterMessageListener;
import com.whz.mlc.cache.stat.MlcStatManager;
import com.whz.mlc.cache.stat.impl.RedisMlcStatManager;
import com.whz.mlc.cache.local.event.listener.GlobalEventListener;
import com.whz.mlc.redis.LogErrorHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;

import java.util.concurrent.ExecutorService;

@Configuration
@ConditionalOnBean(RedisConnectionFactory.class)
public class MlcRedisMessageConfiguration {

    @Value("${mlc.message.queue.task.size:500}")
    public int queueSize;
    @Value("${mlc.message.queue.sub.size:500}")
    public int subSize;
    @Value("${mlc.message.queue.keepalive.ms:30000}")
    public int keepAliveMs;

    /**
     * 创建二级缓存客户端
     */
    @Bean
    @ConditionalOnMissingBean
    public MlcRedisClient createRedisClient(RedisConnectionFactory redisConnectionFactory) {
        return new MlcRedisClient(redisConnectionFactory);
    }

    @Bean
    @ConditionalOnMissingBean
    public MlcStatManager createTmcStatManager(MlcRedisClient tmcRedisClient) {
        return new RedisMlcStatManager(tmcRedisClient.getRedisTemplate());
    }

    /**
     * 集群事件监听
     */
    @Bean
    @Primary
    public RedisMessageListenerContainer tmcMessageListenerContainer(
            @Autowired(required = false) RedisConnectionFactory redisConnectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(redisConnectionFactory);
        container.setTaskExecutor(redisTaskExecutor());
        container.setSubscriptionExecutor(redisSubscriptionExecutor());
        container.setErrorHandler(new LogErrorHandler());
        container.addMessageListener(new ClusterMessageListener(), ChannelTopic.of(MlcConstants.REDIS_TOPIC));
        return container;
    }

    @Bean
    @Primary
    public GlobalEventListener createCacheEventGlobalListener() {
        return new GlobalEventListener();
    }

    private ExecutorService redisTaskExecutor() {
        return DefaultCommonThreadPool.DEFAULT_THREAD_POOL
                .defaultAsyncExecutorService("mlc-redis-task",
                                             1,
                                             1,
                                             keepAliveMs,
                                             queueSize);
    }

    private ExecutorService redisSubscriptionExecutor() {
        return DefaultCommonThreadPool.DEFAULT_THREAD_POOL
                .defaultAsyncExecutorService("mlc-redis-sub",
                                             1,
                                             1,
                                             keepAliveMs,
                                             subSize);
    }

}